package cn.itcast.streaming.function.window;

import cn.itcast.streaming.bean.ItcastDataPartObj;
import cn.itcast.streaming.bean.OnlineDataObj;
import cn.itcast.streaming.utils.DateUtil;
import com.google.common.collect.Lists;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.LinkedList;

//自定义窗口函数开发
public class OnlineStatisticsWindowFunction implements WindowFunction<ItcastDataPartObj, OnlineDataObj, String, TimeWindow> {
    @Override
    public void apply(String s, TimeWindow timeWindow, Iterable<ItcastDataPartObj> iterable, Collector<OnlineDataObj> collector) throws Exception {
        //todo 1.将迭代器中的数据转换成本地集合对象
        LinkedList<ItcastDataPartObj> itcastDataPartObjs = Lists.newLinkedList(iterable);
        //todo 2.对集合对象的所有数据按照事件时间进行排序
        itcastDataPartObjs.sort((o1, o2) -> {
            if(o1.getTerminalTimeStamp()>o2.getTerminalTimeStamp()){
                return 1;
            }else if(o1.getTerminalTimeStamp()<o2.getTerminalTimeStamp()){
                return -1;
            }else {
                return 0;
            }
        });
        //todo 3.循环遍历每条数据，过滤出异常指标的数据进行拼接到指定属性中存储
        ItcastDataPartObj firstObj = itcastDataPartObjs.getFirst();
        for (ItcastDataPartObj obj : itcastDataPartObjs) {
            //判断当前数据中是否存在异常的数据
            if(filterNoAlarm(obj)){
                //没有报警字段，对javabean进行拉宽操作
                OnlineDataObj onlineDataObj = setOnlineDataObj(obj, firstObj, 0);
                System.out.println("不存在异常指标数据");
                collector.collect(onlineDataObj);
            }else {
                //如果有报警字段，对javabean继续拉宽操作
                OnlineDataObj onlineDataObj = setOnlineDataObj(obj, firstObj, 1);
                System.out.println("存在异常指标数据");
                collector.collect(onlineDataObj);
            }
        } 
    }

    //对原始数据进行拉宽操作
    private OnlineDataObj setOnlineDataObj(ItcastDataPartObj obj, ItcastDataPartObj firstObj, int isAlarm) throws InvocationTargetException, IllegalAccessException {
        //定义需要返回的javabean对象

        OnlineDataObj onlineDataObj = new OnlineDataObj();
        /*
        onlineDataObj.setVin(obj.getVin());
        onlineDataObj.setLat(obj.getLat());
        onlineDataObj.setLng(obj.getLng());
        onlineDataObj.setMileage(obj.getTotalOdometer());
        onlineDataObj.setIsAlarm(isAlarm);
        onlineDataObj.setAlarmName(String.join("~",addAlarmNameList(obj)));
        onlineDataObj.setTerminalTime(obj.getTerminalTime());
        onlineDataObj.setEarliestTime(firstObj.getTerminalTime());
        onlineDataObj.setMaxVoltageBattery(obj.getMaxVoltageBattery());
        onlineDataObj.setMinVoltageBattery(obj.getMinVoltageBattery());
        onlineDataObj.setMaxTemperatureValue(obj.getMaxTemperatureValue());
        onlineDataObj.setMinTemperatureValue(obj.getMinTemperatureValue());
        onlineDataObj.setSpeed(obj.getSpeed());
        onlineDataObj.setSoc(obj.getSoc());
        onlineDataObj.setChargeFlag(obj.getChargeStatus()); //充电标识
        onlineDataObj.setTotalVoltage(obj.getTotalVoltage());
        onlineDataObj.setTotalCurrent(obj.getTotalCurrent());
        onlineDataObj.setBatteryVoltage(obj.getBatteryVoltage());
        onlineDataObj.setProbeTemperatures(obj.getProbeTemperatures());
        onlineDataObj.setProcessTime(DateUtil.getCurrentTime());
        onlineDataObj.setProvince(obj.getProvince());
        onlineDataObj.setCity(obj.getCity());
        onlineDataObj.setDistrict(obj.getDistrict());*/

        //将源端的属性赋值给目标端bean对象的同名方法
        BeanUtils.copyProperties(onlineDataObj,obj);
        onlineDataObj.setMileage(obj.getTotalOdometer());
        onlineDataObj.setIsAlarm(isAlarm);
        onlineDataObj.setAlarmName(String.join("~",addAlarmNameList(obj)));
        onlineDataObj.setChargeFlag(obj.getChargeStatus()); //充电标识
        onlineDataObj.setProcessTime(DateUtil.getCurrentTime());

        return onlineDataObj;
    }

    private ArrayList<String> addAlarmNameList(ItcastDataPartObj itcastDataPartObj){
        //定义故障名称列表对象
        ArrayList<String> alarmNameList = new ArrayList<>();
        //电池高温报警
        if(itcastDataPartObj.getBatteryAlarm() ==1){alarmNameList.add("电池高温报警");}
        if(itcastDataPartObj.getSingleBatteryOverVoltageAlarm()==1){alarmNameList.add("电池单体一致性差报警");}
        if(itcastDataPartObj.getBatteryConsistencyDifferenceAlarm()==1){alarmNameList.add("电池单体一致性差报警");}
        if(itcastDataPartObj.getInsulationAlarm()==1){alarmNameList.add("绝缘报警");}
        if(itcastDataPartObj.getHighVoltageInterlockStateAlarm()==1){alarmNameList.add("高压互锁状态报警");}
        if(itcastDataPartObj.getSocJumpAlarm()==1){alarmNameList.add("SOC跳变报警");}
        if(itcastDataPartObj.getDriveMotorControllerTemperatureAlarm()==1){alarmNameList.add("驱动电机控制器温度报警");}
        if(itcastDataPartObj.getDcdcTemperatureAlarm()==1){alarmNameList.add("DC-DC温度报警");}
        if(itcastDataPartObj.getSocHighAlarm()==1){alarmNameList.add("SOC过高报警");}
        if(itcastDataPartObj.getSocLowAlarm()==1){alarmNameList.add("SOC低报警");}
        if(itcastDataPartObj.getTemperatureDifferenceAlarm()==1){alarmNameList.add("温度差异报警");}
        if(itcastDataPartObj.getVehicleStorageDeviceUndervoltageAlarm()==1){alarmNameList.add("车载储能装置欠压报警");}
        if(itcastDataPartObj.getDcdcStatusAlarm()==1){alarmNameList.add("DC状态报警");}
        if(itcastDataPartObj.getSingleBatteryUnderVoltageAlarm()==1){alarmNameList.add("单体电池欠压报警");}
        if(itcastDataPartObj.getRechargeableStorageDeviceMismatchAlarm()==1){alarmNameList.add("可充电储能系统不匹配报警");}
        if(itcastDataPartObj.getVehicleStorageDeviceOvervoltageAlarm()==1){alarmNameList.add("车载储能装置过压报警");}
        if(itcastDataPartObj.getBrakeSystemAlarm()==1){alarmNameList.add("制动系统报警");}
        if(itcastDataPartObj.getDriveMotorTemperatureAlarm()==1){alarmNameList.add("驱动电机温度报警");}
        if(itcastDataPartObj.getVehiclePureDeviceTypeOvercharge()==1){alarmNameList.add("车载储能装置类型过充报警");}
        return alarmNameList;
    }


    /**
     * 判断数据中是否包含了异常字段信息
     * @param obj
     * @return
     */
    private boolean filterNoAlarm(ItcastDataPartObj obj) {

        if(
                obj.getBatteryAlarm()==1 ||  //单体电池高压报警
                obj.getSingleBatteryOverVoltageAlarm()==1 ||//电池单体一致性差报警
                obj.getBatteryConsistencyDifferenceAlarm()==1 ||// 电池单体一致性差报警
                obj.getInsulationAlarm()==1 || //绝缘报警
                obj.getHighVoltageInterlockStateAlarm()==1 ||//高压互锁状态报警
                obj.getSocJumpAlarm()==1 || //SOC跳变报警
                obj.getDriveMotorControllerTemperatureAlarm()==1 || //驱动电机控制器温度报警
                obj.getDcdcTemperatureAlarm()==1 || //DC-DC温度报警（dc-dc可以理解为车辆动力智能系统转换器）
                obj.getSocHighAlarm()==1 ||  //SOC过高报警
                obj.getSocLowAlarm()==1 ||  //SOC低报警
                obj.getTemperatureDifferenceAlarm()==1 ||  //温度差异报警
                obj.getVehicleStorageDeviceUndervoltageAlarm()==1 ||  //车载储能装置欠压报警
                obj.getDcdcStatusAlarm()==1 ||  //DC-DC状态报警
                obj.getSingleBatteryUnderVoltageAlarm()==1 ||  //单体电池欠压报警
                obj.getRechargeableStorageDeviceMismatchAlarm()==1 ||  //可充电储能系统不匹配报警
                obj.getVehicleStorageDeviceOvervoltageAlarm()==1 ||  //车载储能装置过压报警
                obj.getBrakeSystemAlarm()==1 ||  //制动系统报警
                obj.getDriveMotorTemperatureAlarm()==1 ||  //驱动电机温度报警
                obj.getVehiclePureDeviceTypeOvercharge()==1  //车载储能装置类型过充报警
           ){
            return false;
        }else {
            return true;
        }
    }
}
